package com.wjk.kylin.rpc.core.netty.server;

import com.wjk.kylin.rpc.core.netty.AbstractServer;
import com.wjk.kylin.rpc.core.provider.RpcProviderFactory;
import com.wjk.kylin.rpc.core.util.ThreadPoolUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.util.concurrent.ThreadPoolExecutor;

/**
 *
 * @author wjk
 */
public class NettyServer extends AbstractServer {
    private Thread thread;

    /**
     * 启动服务
     *
     * @param providerFactory
     * @throws Exception
     */
    @Override
    public void start(RpcProviderFactory providerFactory) throws Exception {
        thread = new Thread(() -> {
            // param
            final ThreadPoolExecutor serverHandlerPool = ThreadPoolUtil.makeServerThreadPool(
                    NettyServer.class.getSimpleName(), providerFactory.getProperties().getCorePoolSize(), providerFactory.getProperties().getMaxPoolSize());
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();

            try {
                // start server
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel 作为服务器的通道实现
                        .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接 // childOption SocketChannel
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childHandler(new NettyServerChannelInitializer(providerFactory, serverHandlerPool));//给workerGroup EventLoopGroup对应的管道设置处理器

                // bind
                ChannelFuture future = bootstrap.bind(providerFactory.getPort()).sync();

                LOGGER.info(">>>>>>>>>>> kylin-rpc netty server start success, nettyServerName = {}, port = {}", NettyServer.class.getName(), providerFactory.getPort());
                onStart();

                // wait util stop
                future.channel().closeFuture().sync();

            } catch (Exception e) {
                if (e instanceof InterruptedException) {
                    LOGGER.info(">>>>>>>>>>>> kylin-rpc remoting server stop");
                } else {
                    LOGGER.error(">>>>>>>>>>> kylin-rpc remoting server error", e);
                }
            } finally {
                // stop
                try {
                    serverHandlerPool.shutdown();    // shutdownNow
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
                try {
                    workerGroup.shutdownGracefully();
                    bossGroup.shutdownGracefully();
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
        });
        //设置为守护线程
        thread.setDaemon(true);
        thread.start();
    }

    @Override
    public void stop() throws Exception {
        //destroy server thread
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }

        // on stop
        onStop();
        LOGGER.info(">>>>>>>>>>> kylin-rpc remoting server destroy success");
    }
}
